package com.atguigu.flink.datastramapi.transform;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/25
 *
 *  Connect: 联合，可以将两个不同类型的流，联合，进行处理。
 *
 *  最低成本卖牛肉丸？
 *
 *    猪肉(9) ---->  搅碎，添加牛肉精 ----->  牛肉味的肉馅
 *      T     ----->process(T t)--------------------------------> X
 *
 *    牛肉馅(1) ----> 搅碎   -------------> 牛肉味的肉馅
 *       E  ------> process(E e)--------------------------------> x
 *
 *
 */
public class Demo11_Connect
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<Pork> ds1 = env.fromElements(new Pork(), new Pork());
        DataStreamSource<Beef> ds2 = env.fromElements(new Beef(), new Beef());

        ds1.connect(ds2)
           .process(new CoProcessFunction<Pork, Beef, String>()
           {
               private String operate = "搅碎";

               private Pork p;
               //处理第一个流中的数据
               @Override
               public void processElement1(Pork value, Context ctx, Collector<String> out) throws Exception {

                   p = value;
                   System.out.println(value.getType() + "添加牛肉精, 被" + operate);
                   out.collect("牛肉味的肉馅");

               }

               //处理第二个流中的数据
               @Override
               public void processElement2(Beef value, Context ctx, Collector<String> out) throws Exception {
                   System.out.println(p);
                   System.out.println(value.getType() + "被" + operate);
                   out.collect("牛肉味的肉馅");
               }
           })
           .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    public static class Pork{
        private String type ="猪肉";
    }

    @AllArgsConstructor
    @NoArgsConstructor
    @Data
    public static class Beef{
        private String type ="牛肉";
    }
}
